package org.apache.rocketmq.store.index;



import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.logfile.DefaultMappedFile;
import org.apache.rocketmq.store.logfile.MappedFile;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.util.List;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/7/14
 * @方法描述：索引文件实现类
 */
public class IndexFile {

    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    //每个哈希槽的大小
    private static int hashSlotSize = 4;

    //消息索引的大小，实际上就是索引文件存储消息key的存储单元大小
    //一共是20个字节
    //┌───────────────┬───────────────────────────────┬───────────────┬───────────────┐
    //│ Key HashCode  │ Physical Offset   │ Time Diff │ Next Index Pos│
    //│ (4 Bytes)     │ (8 Bytes)         │ (4 Bytes) │ (4 Bytes) │
    //├───────────────┴───────────────────────────────┴───────────────┴───────────────┤
    //从上图可以看到，前4个字节存储的是消息的key的哈希值，后面8个字节是这条消息在CommitLog文件中的全局偏移量，后面4个字节是这条消息
    //存储的时间和索引文件中第一条消息存储时间的差值，如果差值小于0则意味着该索引条目无效
    //最后4个字节存储的是哈希槽中上一条消息的索引，现在大家可能还不清楚这一切是什么意思，别着急，往下看到putKey方法就全明白了
    private static int indexSize = 20;

    //哈希槽存储的无效索引
    private static int invalidIndex = 0;

    //哈希槽总数量，默认是5000000，也就是五百万个
    private final int hashSlotNum;

    //索引文件可以存储的索引条目数量，默认是2千万个
    private final int indexNum;

    //当前索引文件的容量，这个容量就等于索引文件头大小+所有哈希槽大小+存储的所有索引条目大小
    private final int fileTotalSize;

    //内存映射文件
    private final MappedFile mappedFile;

    //内存映射缓冲区
    private final MappedByteBuffer mappedByteBuffer;

    //索引文件头
    private final IndexHeader indexHeader;


    //构造方法
    public IndexFile(final String fileName, final int hashSlotNum, final int indexNum,
                     final long endPhyOffset, final long endTimestamp) throws IOException {
        //在这里计算索引文件的总大小，总大小就是文件头大小+所有哈希槽大小+存储的所有索引条目大小
        this.fileTotalSize = IndexHeader.INDEX_HEADER_SIZE + (hashSlotNum * hashSlotSize) + (indexNum * indexSize);
        //创建内存映射文件
        this.mappedFile = new DefaultMappedFile(fileName, fileTotalSize);
        //得到内存映射缓冲区
        this.mappedByteBuffer = this.mappedFile.getMappedByteBuffer();
        //得到哈希槽数量
        this.hashSlotNum = hashSlotNum;
        //得到索引条目总数量
        this.indexNum = indexNum;
        ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
        //创建文件头，在刷新数据的时候会先把文件头的数据刷新到文件中，然后才是后面的所有数据
        this.indexHeader = new IndexHeader(byteBuffer);
        if (endPhyOffset > 0) {
            //在创建新的索引文件时，如果这个索引文件并不是文件队列中的第一个文件，则需要获取前一个索引文件的末尾物理偏移量
            //因为索引文件存储的消息的物理偏移量是递增的，所以当前索引文件的开始物理偏移量就是前一个索引文件的结束物理偏移量
            //所谓开始偏移量就是当前索引文件存储的第一条消息在CommitLog文件中的全局偏移量
            this.indexHeader.setBeginPhyOffset(endPhyOffset);
            this.indexHeader.setEndPhyOffset(endPhyOffset);
        }
        if (endTimestamp > 0) {
            //设置新的索引文件的开始存储消息的时间戳，大家可以看到，不管是上面的物理偏移量还是存储消息时间戳
            //都是设置到文件头中，这就表示索引文件头中持有了索引文件的元数据
            this.indexHeader.setBeginTimestamp(endTimestamp);
            this.indexHeader.setEndTimestamp(endTimestamp);
        }
    }


    public String getFileName() {
        return this.mappedFile.getFileName();
    }


    public int getFileSize() {
        return this.fileTotalSize;
    }


    public void load() {
        this.indexHeader.load();
    }


    //终止服务的方法
    public void shutdown() {
        //这里可以看到，在关闭索引服务之前会把索引文件的数据刷新到硬盘
        this.flush();
        UtilAll.cleanBuffer(this.mappedByteBuffer);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：刷新索引文件数据到本地硬盘的操作
     */
    public void flush() {
        long beginTime = System.currentTimeMillis();
        //得到资源锁
        if (this.mappedFile.hold()) {
            //先把文件头的数据更新到索引文件缓冲区中
            this.indexHeader.updateByteBuffer();
            //在这里执行刷盘操作
            this.mappedByteBuffer.force();
            //释放资源锁
            this.mappedFile.release();
            log.info("flush index file elapsed time(ms) " + (System.currentTimeMillis() - beginTime));
        }
    }


    //判断索引文件是否写满的方法
    public boolean isWriteFull() {
        return this.indexHeader.getIndexCount() >= this.indexNum;
    }


    //销毁索引文件的方法
    public boolean destroy(final long intervalForcibly) {
        return this.mappedFile.destroy(intervalForcibly);
    }




    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向索引文件存放数据的方法，方法参数key表示用户为消息定义的key，phyOffset表示消息在在CommitLog文件中的全局偏移量，storeTimestamp表示消息的存储时间戳
     */
    public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
        //判断当前索引文件是否已经写满了，当前文件存储的消息条目数量就存储在文件头中
        //如果没写满，才执行下面的操作，写满就返回false
        if (this.indexHeader.getIndexCount() < this.indexNum) {
            //计算消息key的哈希值
            int keyHash = indexKeyHashMethod(key);
            //哈希值和哈希槽总数取模得到要存储消息索引的哈希槽的位置，所谓消息索引就是要存储的这条消息在索引文件的索引区中的位置
            int slotPos = keyHash % this.hashSlotNum;
            //slotPos * hashSlotSize得到的是刚才定位到的哈希槽索引在哈希槽区的偏移量
            //再加上文件头的大小就得到了哈希槽位置在索引文件中的具体偏移量
            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
            try {//根据这个偏移量从缓冲区中得到这个哈希槽目前存储的文件索引
                //这里我要解释一下，假如有两个消息的key的哈希值一样了，这就意味着发生了哈希冲突，那么就会以链表的方式来存储数据
                //大家想象一下hashmap存储数据的结构即可，这里就会根据得到的值判断当前哈希槽是否存储了有效数据
                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()) {
                    //如果存储的数据无效则把哈希槽置置为无效
                    //如果有效那么刚才得到的slotValue就是之前已经存储在索引文件的消息的索引值
                    slotValue = invalidIndex;
                }//计算消息存储时间与索引文件开始存储消息时间的差值
                long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();
                //转换为毫秒
                timeDiff = timeDiff / 1000;
                //调整时间差边界
                if (this.indexHeader.getBeginTimestamp() <= 0) {
                    timeDiff = 0;
                } else if (timeDiff > Integer.MAX_VALUE) {
                    timeDiff = Integer.MAX_VALUE;
                } else if (timeDiff < 0) {
                    timeDiff = 0;
                }
                //这里就是文件头大小+哈希槽总大小+当前已经存储的索引条目的总大小，这里得到的就是要存储的消息索引条目在索引文件中的起始偏移量
                int absIndexPos = IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize + this.indexHeader.getIndexCount() * indexSize;
                //从起始偏移量开始存储数据，先把消息key的哈希值写入缓冲区
                this.mappedByteBuffer.putInt(absIndexPos, keyHash);
                //再写入消息在CommitLog文件中的全局偏移量
                this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
                //再写入消息的存储时间差
                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
                //这里一步非常关键，这里把哈希槽存储的上一条消息的索引位置写入了当前消息条目中
                //这一以来，只要根据哈希槽存储的现在的消息索引，就能定位到现在要查询的消息，而找到了现在要查询的条目，就可以根据最后四个字节
                //定位到key哈希值一致的上一条消息的索引位置，就可以找到上一条key哈希值相同的消息，这样一来就构成了一个完整的链表
                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
                //在这里把当前消息定位到的哈希槽的值更新了，更新得值就是当前消息在索引文件中的索引，也就是当前消息在索引文件的索引区是第几条消息的信息
                this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
                if (this.indexHeader.getIndexCount() <= 1) {
                    //如果目前要写入消息的索引文件中还没有数据，则将索引条目的起始物理偏移量设置为当前消息的CommitLog文件的全局偏移量
                    //注意，因为indexCount初始化就是1，所以当indexHeader.getIndexCount() <= 1，说明索引文件中并没有数据
                    this.indexHeader.setBeginPhyOffset(phyOffset);
                    //消息存储的起始时间也设为当前消息的存储时间
                    this.indexHeader.setBeginTimestamp(storeTimestamp);
                }
                if (invalidIndex == slotValue) {
                    //如果哈希槽值为空，增加哈希槽使用数
                    this.indexHeader.incHashSlotCount();
                }//在这里增加文件头记录的索引文件存储的消息条目数
                this.indexHeader.incIndexCount();
                //把当前存储消息在CommitLog文件的全局偏移量设置到文件头中
                this.indexHeader.setEndPhyOffset(phyOffset);
                //更新文件头最新消息的存储时间
                this.indexHeader.setEndTimestamp(storeTimestamp);
                return true;
            } catch (Exception e) {
                log.error("putKey exception, Key: " + key + " KeyHashCode: " + key.hashCode(), e);
            }
        } else {
            log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()
                    + "; index max num = " + this.indexNum);
        }
        return false;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：计算key的哈希值的方法
     */
    public int indexKeyHashMethod(final String key) {
        int keyHash = key.hashCode();
        int keyHashPositive = Math.abs(keyHash);
        if (keyHashPositive < 0) {
            keyHashPositive = 0;
        }
        return keyHashPositive;
    }


    public long getBeginTimestamp() {
        return this.indexHeader.getBeginTimestamp();
    }


    public long getEndTimestamp() {
        return this.indexHeader.getEndTimestamp();
    }


    public long getEndPhyOffset() {
        return this.indexHeader.getEndPhyOffset();
    }


    //判断指定时间范围内是否有索引条目的方法
    public boolean isTimeMatched(final long begin, final long end) {
        boolean result = begin < this.indexHeader.getBeginTimestamp() && end > this.indexHeader.getEndTimestamp();
        result = result || (begin >= this.indexHeader.getBeginTimestamp() && begin <= this.indexHeader.getEndTimestamp());
        result = result || (end >= this.indexHeader.getBeginTimestamp() && end <= this.indexHeader.getEndTimestamp());
        return result;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：从索引文件中查询指定key的方法，该方法会返回一个集合，集合中存储的就是相同key的消息在的CommitLog文件中的全局偏移量
     */
    public void selectPhyOffset(final List<Long> phyOffsets, final String key, final int maxNum, final long begin, final long end) {
        //得到文件资源锁
        if (this.mappedFile.hold()) {
            //计算key的哈希值
            int keyHash = indexKeyHashMethod(key);
            //得到对应的哈希槽位置
            int slotPos = keyHash % this.hashSlotNum;
            //得到哈希槽在索引文件中的偏移量
            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
            try {//得到哈希槽位置存储的索引值，这个索引值就是要查询的消息在索引文件索引区的索引
                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                //判断刚才得到的索引值是否有效
                if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()
                        || this.indexHeader.getIndexCount() <= 1) {
                } else {//有效则开始获取对应的消息，这里之所以循环获取，是考虑有链表的情况，即一个key对应多个索引值
                    for (int nextIndexToRead = slotValue; ; ) {
                        //如果找到的条目数超过了阈值，则退出循环，不再继续寻找对应的消息了
                        if (phyOffsets.size() >= maxNum) {
                            break;
                        }//得到当前要查找的消息在索引文件中的偏移量
                        int absIndexPos = IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize + nextIndexToRead * indexSize;
                        //根据偏移量的到消息key的哈希值
                        int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
                        //得到消息在CommitLog文件中的全局偏移量
                        long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);
                        //得到消息存储的时间差
                        long timeDiff = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
                        //得到上一个key的哈希值相同的消息索引位置
                        int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);
                        if (timeDiff < 0) {
                            //时间差小于0则退出循环
                            break;
                        }
                        timeDiff *= 1000L;
                        //计算当前查找到的消息的实际存储时间
                        long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
                        //判断消息存储时间是否在指定的时间范围之内
                        boolean timeMatched = timeRead >= begin && timeRead <= end;
                        //如果符合要求则把消息在CommitLog文件中的全局偏移量添加到结果集中
                        if (keyHash == keyHashRead && timeMatched) {
                            phyOffsets.add(phyOffsetRead);
                        }//然后判断当前消息记录的上一条消息的索引是否有效
                        if (prevIndexRead <= invalidIndex
                                || prevIndexRead > this.indexHeader.getIndexCount()
                                || prevIndexRead == nextIndexToRead || timeRead < begin) {
                            break;
                        }//有效则更新索引位置，进入下一次循环，查找下一条消息
                        nextIndexToRead = prevIndexRead;
                    }
                }
            } catch (Exception e) {
                log.error("selectPhyOffset exception ", e);
            } finally {
                this.mappedFile.release();
            }
        }
    }
}

